package com.askyjt.videosearch.service.impl;

import com.askyjt.videosearch.bo.SocketDTO;
import com.askyjt.videosearch.common.constant.Constant;
import com.askyjt.videosearch.common.constant.MilvusParamEnum;
import com.askyjt.videosearch.config.MilvusConfig;
import com.askyjt.videosearch.entity.KeyFrame;
import com.askyjt.videosearch.entity.Video;
import com.askyjt.videosearch.mapper.VideoMapper;
import com.askyjt.videosearch.service.KeyFrameService;
import com.askyjt.videosearch.service.VideoService;
import com.askyjt.videosearch.util.Result;
import com.askyjt.videosearch.util.SocketUtil;
import com.askyjt.videosearch.vo.VideoVo;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import io.milvus.client.ConnectFailedException;
import io.milvus.client.MilvusClient;
import io.milvus.client.Response;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author yangjintao
 * @since 2020-11-02
 */
@Slf4j
@Service
public class VideoServiceImpl extends ServiceImpl<VideoMapper, Video> implements VideoService {

    @Resource
    private KeyFrameService keyFrameService;

    @Resource
    private VideoMapper videoMapper;

    @Resource
    private MilvusConfig milvusConfig;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean saveVideo(Video video) {
        //封装请求socket的对象
        HashMap<String, String> parameters = new HashMap<>(8);
        parameters.put(MilvusParamEnum.VIDEO_PATH.getParamName(), video.getVideoUri());
        parameters.put(MilvusParamEnum.VIDEO_NAME.getParamName(), video.getVideoName());
        parameters.put(MilvusParamEnum.TABLE_NAME.getParamName(), Constant.MILVUS_KEYFRAME_TABLE);
        SocketDTO socketDTO = SocketDTO.builder().method(Constant.SAVE_VIDEO_TO_MILVUS).parameter(parameters).build();
        //请求特征提取相关socket
        Result<Map<String, List<String>>> result = SocketUtil.remoteCallAdd(socketDTO);
        if (result != null && result.getCode() == Result.SUCCESS) {
            List<Long> milvusIds = result.getData().get(MilvusParamEnum.MILVUS_IDS.getParamName())
                    .stream().map(Long::valueOf).collect(Collectors.toList());
            List<String> url = result.getData().get(MilvusParamEnum.URL.getParamName());
            List<Long> durationTimes = result.getData().get(MilvusParamEnum.DURATION_TIME.getParamName())
                    .stream().map(Long::valueOf).collect(Collectors.toList());
            Long duration = Long.valueOf(result.getData().get(MilvusParamEnum.DURATION.getParamName()).get(0));
            //插入数据库
            video.setDuration(duration);
            boolean insertVideo = video.insert();
            List<KeyFrame> keyframes = new ArrayList<>();
            for (int i = 0; i < milvusIds.size(); i++) {
                keyframes.add(new KeyFrame()
                        .setMilvusId(milvusIds.get(i))
                        .setKeyPosition(durationTimes.get(i))
                        .setUrl(url.get(i))
                        .setVideoId(video.getId()));
            }
            keyframes.forEach(System.out::println);
            boolean saveResult = keyFrameService.saveBatch(keyframes);
            return saveResult && insertVideo;
        }
        return false;
    }

    @Override
    public List<VideoVo> searchVideo(String path) {
        //封装用于请求的对象
        Map<String, String> parameters = new HashMap<>(8);
        parameters.put(MilvusParamEnum.IMG_PATH.getParamName(), path);
        parameters.put(MilvusParamEnum.TABLE_NAME.getParamName(), Constant.MILVUS_KEYFRAME_TABLE);
        parameters.put(MilvusParamEnum.TOP_K.getParamName(), "10");
        //调用socket搜索数据。
        SocketDTO socketDTO = SocketDTO.builder().method(Constant.SEARCH_VIDEO_OR_IMG).parameter(parameters).build();
        Result<Map<String, List<String>>> result = SocketUtil.remoteCallSearch(socketDTO);
        AtomicInteger index = new AtomicInteger(-1);
        //解析收到的数据并进行封装
        if (result != null && result.getCode() == Result.SUCCESS) {
            List<Long> milvusIds = result.getData().get(MilvusParamEnum.MILVUS_IDS.getParamName())
                    .stream().map(Long::valueOf).collect(Collectors.toList());
            List<String> disList = result.getData().get(MilvusParamEnum.MILVUS_DISLIST.getParamName());
            HashMap<Long, Double> milvusMap = new HashMap<>(8);
            milvusIds.forEach(mId -> milvusMap.put(mId, Double.valueOf(disList.get(index.incrementAndGet()))));
            List<KeyFrame> keyFrames = keyFrameService.list(new QueryWrapper<KeyFrame>().in(KeyFrame.MILVUS_ID, milvusIds));
            if (CollectionUtils.isEmpty(keyFrames)) {
                return null;
            }
            keyFrames.forEach(x -> x.setDistance(milvusMap.get(x.getMilvusId())));
            List<Integer> videoIds = keyFrames
                    .stream().sorted(Comparator.comparingDouble(KeyFrame::getDistance).reversed())
                    .map(KeyFrame::getVideoId)
                    .distinct()
                    .collect(Collectors.toList());
            List<VideoVo> videoVos = videoMapper.getVideoVoList(videoIds,milvusIds);
            //去除重复时间点的数据
            if (CollectionUtils.isNotEmpty(videoVos)) {
                List<VideoVo> videoVoList = videoVos
                        .stream().collect(Collectors.collectingAndThen(
                                Collectors.toCollection(() -> new TreeSet<>(
                                        Comparator.comparing(VideoVo::getKeyPosition))), ArrayList::new));
                //根据distance排序并返回
                return videoVoList.stream().map(videoVo -> videoVo.setDistance(milvusMap.get(videoVo.getMilvusId())))
                        .sorted(Comparator.comparingDouble(VideoVo::getDistance)).collect(Collectors.toList());
            }
        }
        return null;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean deleteVideoById(List<Long> ids) throws InterruptedException {
        int deleteResult = videoMapper.deleteBatchIds(ids);
        if (deleteResult > 0) {
            keyFrameService.remove(new QueryWrapper<KeyFrame>().in(KeyFrame.VIDEO_ID, ids));
            MilvusClient client = null;
            try {
                client = milvusConfig.getMilvusClient();
                Response response = client.deleteEntityByID(Constant.MILVUS_PICTURE_TABLE, ids);
                log.info("删除milvus向量结果： code：" + response.getStatus().getCode() + " message" + response.getMessage());
            } catch (ConnectFailedException e) {
                e.printStackTrace();
            }finally {
                if (client != null) {
                    client.disconnect();
                }
            }
        }
        return false;
    }


}
